package beautiful.butterfly.drds.data_exchange.writer.util;


import beautiful.butterfly.drds.data_exchange.exception.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.util.*;
import beautiful.butterfly.drds.data_exchange.writer.Constant;
import beautiful.butterfly.drds.data_exchange.writer.Key;
import com.alibaba.druid.sql.parser.ParserException;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public final class WriterUtil
{
    private static final Logger LOG = LoggerFactory.getLogger(WriterUtil.class);

    //TODO 切分报错
    public static List<Configuration> doSplit(Configuration simplifiedConf,
                                              int adviceNumber)
    {

        List<Configuration> splitResultConfigs = new ArrayList<Configuration>();

        int tableNumber = simplifiedConf.getInt(Constant.table_number);

        //处理单表的情况
        if (tableNumber == 1)
        {
            //由于在之前的  master prepare 中已经把 tableName,jdbcUrl 提取出来，所以这里处理十分简单
            for (int j = 0; j < adviceNumber; j++)
            {
                splitResultConfigs.add(simplifiedConf.clone());
            }

            return splitResultConfigs;
        }

        if (tableNumber != adviceNumber)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.CONF_ERROR,
                    String.format("您的配置文件中的列配置信息有误. 您要写入的目的端的表个数是:%s , 但是根据系统建议需要切分的份数是：%s. 请检查您的配置并作出修改.",
                            tableNumber, adviceNumber));
        }

        String jdbcUrl;
        List<String> preSqls = simplifiedConf.getList(Key.pre_sql, String.class);
        List<String> postSqls = simplifiedConf.getList(Key.post_sql, String.class);

        List<Object> conns = simplifiedConf.getList(Constant.connection,
                Object.class);

        for (Object conn : conns)
        {
            Configuration sliceConfig = simplifiedConf.clone();

            Configuration connConf = Configuration.from(conn.toString());
            jdbcUrl = connConf.getString(Key.jdbc_url);
            sliceConfig.set(Key.jdbc_url, jdbcUrl);

            sliceConfig.remove(Constant.connection);

            List<String> tables = connConf.getList(Key.table_name, String.class);

            for (String table : tables)
            {
                Configuration tempSlice = sliceConfig.clone();
                tempSlice.set(Key.table_name, table);
                tempSlice.set(Key.pre_sql, renderPreOrPostSqls(preSqls, table));
                tempSlice.set(Key.post_sql, renderPreOrPostSqls(postSqls, table));

                splitResultConfigs.add(tempSlice);
            }

        }

        return splitResultConfigs;
    }

    public static List<String> renderPreOrPostSqls(List<String> preOrPostSqls, String tableName)
    {
        if (null == preOrPostSqls)
        {
            return Collections.emptyList();
        }

        List<String> renderedSqls = new ArrayList<String>();
        for (String sql : preOrPostSqls)
        {
            //preSql为空时，不加入执行队列
            if (StringUtils.isNotBlank(sql))
            {
                renderedSqls.add(sql.replace(Constant.table_name_placeholder, tableName));
            }
        }

        return renderedSqls;
    }

    public static void executeSqls(Connection connection, List<String> sqlList, String basicMessage)
    {
        Statement statement = null;
        String currentSql = null;
        try
        {
            statement = connection.createStatement();
            for (String sql : sqlList)
            {
                currentSql = sql;
                Jdbc.execute(statement, sql);
            }
        } catch (Exception e)
        {
            throw RdbmsException.asQueryException(e, currentSql, null, null);
        } finally
        {
            Jdbc.close(null, statement, null);
        }
    }

    public static String getWriteTemplate(List<String> columnHolders, List<String> valueHolders, String writeMode, boolean forceUseUpdate)
    {
        boolean isWriteModeLegal = writeMode.startsWith("insert") || writeMode.startsWith("replace") || writeMode.startsWith("update");
        if (!isWriteModeLegal)
        {
            throw DataExchangeException.asDataExchangeException(DBUtilErrorCode.ILLEGAL_VALUE,
                    String.format("您所配置的 writeMode:%s 错误. 因为DataX 目前仅支持replace,update 或 insert 方式. 请检查您的配置并作出修改.", writeMode));
        }
        // && writeMode.trim().toLowerCase().startsWith("replace")
        String writeDataSqlTemplate;
        if (forceUseUpdate ||
                writeMode.trim().toLowerCase().startsWith("update"))

        {
            //update只在mysql下使用

            writeDataSqlTemplate = new StringBuilder()
                    .append("insert into %s (").append(StringUtils.join(columnHolders, ","))
                    .append(") values(").append(StringUtils.join(valueHolders, ","))
                    .append(")")
                    .append(onDuplicateKeyUpdateString(columnHolders))
                    .toString();
        } else
        {

            //这里是保护,如果其他错误的使用了update,需要更换为replace
            if (writeMode.trim().toLowerCase().startsWith("update"))
            {
                writeMode = "replace";
            }
            writeDataSqlTemplate = new StringBuilder().append(writeMode)
                    .append(" INTO %s (").append(StringUtils.join(columnHolders, ","))
                    .append(") VALUES(").append(StringUtils.join(valueHolders, ","))
                    .append(")").toString();
        }

        return writeDataSqlTemplate;
    }

    public static String onDuplicateKeyUpdateString(List<String> columnHolders)
    {
        if (columnHolders == null || columnHolders.size() < 1)
        {
            return "";
        }
        StringBuilder sb = new StringBuilder();
        sb.append(" ON DUPLICATE KEY UPDATE ");
        boolean first = true;
        for (String column : columnHolders)
        {
            if (!first)
            {
                sb.append(",");
            } else
            {
                first = false;
            }
            sb.append(column);
            sb.append("=VALUES(");
            sb.append(column);
            sb.append(")");
        }

        return sb.toString();
    }

    public static void preCheckPrePareSQL(Configuration originalConfig)
    {
        List<Object> conns = originalConfig.getList(Constant.connection, Object.class);
        Configuration connConf = Configuration.from(conns.get(0).toString());
        String table = connConf.getList(Key.table_name, String.class).get(0);

        List<String> preSqls = originalConfig.getList(Key.pre_sql,
                String.class);
        List<String> renderedPreSqls = WriterUtil.renderPreOrPostSqls(
                preSqls, table);

        if (null != renderedPreSqls && !renderedPreSqls.isEmpty())
        {
            LOG.info("Begin to preCheck preSqlList:[{}].",
                    StringUtils.join(renderedPreSqls, ";"));
            for (String sql : renderedPreSqls)
            {
                try
                {
                    Jdbcs1.sqlValid(sql);
                } catch (ParserException e)
                {
                    throw RdbmsException.asPreSQLParserException(e, sql);
                }
            }
        }
    }

    public static void preCheckPostSQL(Configuration originalConfig)
    {
        List<Object> conns = originalConfig.getList(Constant.connection, Object.class);
        Configuration connConf = Configuration.from(conns.get(0).toString());
        String table = connConf.getList(Key.table_name, String.class).get(0);

        List<String> postSqls = originalConfig.getList(Key.post_sql,
                String.class);
        List<String> renderedPostSqls = WriterUtil.renderPreOrPostSqls(
                postSqls, table);
        if (null != renderedPostSqls && !renderedPostSqls.isEmpty())
        {

            LOG.info("Begin to preCheck postSqlList:[{}].",
                    StringUtils.join(renderedPostSqls, ";"));
            for (String sql : renderedPostSqls)
            {
                try
                {
                    Jdbcs1.sqlValid(sql);
                } catch (ParserException e)
                {
                    throw RdbmsException.asPostSQLParserException(e, sql);
                }

            }
        }
    }


}
